package cn.whuc.spark.operator

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo_mapPartitions {
  def main(args: Array[String]): Unit = {
    // 1 创建sparkContext
    val sc: SparkContext = new SparkContext(
      new SparkConf()
        .setMaster("local[*]")
        .setAppName(" ")
    )

    // 2 编写代码
    val rdd1: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5), 2)

    val rdd2: RDD[Int] = rdd1.mapPartitions(_.filter(_ % 2 == 0))

//    val rdd2: RDD[Int] = rdd1.mapPartitions(it => {
//      val ints: Iterator[Int] = it.map(_ * 2)
//      println("***")
//      ints
//    })

    rdd2.collect().foreach(println)

    // 对于 map来讲， rdd中的每一个元素都会执行一次map 从rdd执行的全流程上看 所做的工作叫繁杂
    // 对于mapPartitions来讲，rdd中的数据，按分区数执行一次，相当于批处理，效率高
    // mappartitions可能存在的问题，oom
    // 内存大的时候  建议使用 否则 使用map
    // mappartitions可以元素不一一对应



    // 3 关闭上下文对象
    sc.stop()
  }
}
